package com.fdb.efp.basic.framework.mq.listener;

import java.util.List;
import java.util.Map;

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fdb.efp.basic.framework.mq.handle.MQMessageProcessor;

/**
 * @description:接收消息监听处理类(普通监听)
 * @date:2019年2月12日 上午9:24:06
 * @author:wangyk3
 */
public class MessageListenerForConcurrently implements MessageListenerConcurrently {
	private static Logger logger = LoggerFactory.getLogger(MessageListenerForConcurrently.class);
	private Map<String, MQMessageProcessor> handlermap;
	@Override
	public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
		try {
			for (MessageExt msg : msgs) {
				logger.debug(String.format("start consume message: message:id%s topic:%s tags:%s keys:%s message:%s", msg.getMsgId(), msg.getTopic(), msg.getTags(), msg.getKeys(),
						new String(msg.getBody(), "utf-8")));
				MQMessageProcessor handler = handlermap.get(msg.getTopic());
				handler.handleMessage(msg);
				logger.debug(String.format("consume message success: message:id%s topic:%s tags:%s keys:%s message:%s", msg.getMsgId(), msg.getTopic(), msg.getTags(), msg.getKeys(),
						new String(msg.getBody(), "utf-8")));
			}
			// 消费成功
			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
			return ConsumeConcurrentlyStatus.RECONSUME_LATER;
		}
	}
	public void setHandlermap(Map<String, MQMessageProcessor> handlermap) {
		this.handlermap = handlermap;
	}


}
